package db.app

import com.cw.realtime.common.base.FlinkRunner.{SqlRunnerConfig, createBaseDic, createTopicDbTable, runSql}
import com.cw.realtime.common.constant.Constant
import com.cw.realtime.common.util.SqlUtil
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment


object DwdCommentInfo {

  def main(args: Array[String]): Unit = {
    implicit val conf = SqlRunnerConfig("dwd_interaction_comment_info", 10012)

    runSql { (env, tableEnv) =>

      // 将topic_db数据映射到flink sql中
      createTopicDbTable(tableEnv)


      // 从topic_db中提取出comment表
      createCommentInfo(tableEnv)


      // 获取hbase base_dic数据
      createBaseDic(tableEnv)


      // 创建kafka sink表
      createKafkaSinkTable(tableEnv, Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO)

      // look up join
      val joinedTable = joinTable(tableEnv)
      joinedTable.insertInto(Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO).execute()

    }


  }


  def createCommentInfo(tableEnv: StreamTableEnvironment) = {
    val comment_info = tableEnv.sqlQuery(
      """
select
`data`['id'         ] id         ,
`data`['user_id'    ]  user_id    ,
`data`['nick_name'  ]  nick_name  ,
`data`['head_img'   ]  head_img   ,
`data`['sku_id'     ] sku_id    ,
`data`['spu_id'     ]  spu_id     ,
`data`['order_id'   ]  order_id   ,
`data`['appraise'   ]  appraise   ,
`data`['comment_txt' ] comment_txt,
`data`['create_time']  create_time,
`data`['operate_time']  operate_time,
`proc_time`
from topic_db
where `database` = 'gmall'
and `table` = 'comment_info'
and `type` = 'insert'
        """)

    tableEnv.createTemporaryView("comment_info", comment_info)
  }


  def joinTable(tableEnv: StreamTableEnvironment) = {

    tableEnv.sqlQuery(
      """
select
  `id`          ,
  `user_id`     ,
  `nick_name`   ,
  `sku_id`      ,
  `spu_id`      ,
  `order_id`    ,
  c.`appraise` appraise_code   ,
  info.dic_name appraise_name,
  `comment_txt`,
  `create_time`,
  `operate_time`


from comment_info c
left join base_dic FOR SYSTEM_TIME AS OF c.proc_time AS b
on c.appraise = b.rowkey
      """)


  }


  def createKafkaSinkTable(tableEnv: StreamTableEnvironment, topicName: String) = {
    tableEnv.executeSql(
      s"""
create table dwd_interaction_comment_info(

  `id`              STRING,
  `user_id`         STRING,
  `nick_name`       STRING,
  `sku_id`          STRING,
  `spu_id`          STRING,
  `order_id`        STRING,
  `appraise_code`   STRING,
  `appraise_name`   STRING,
  `comment_txt`     STRING,
  `create_time`    STRING,
  `operate_time`    STRING
) ${SqlUtil.getKafkaSinkSql(topicName)}
      """)
  }


}
